package com.runa.monitor.platinterface.netservice.equipment;

import com.alibaba.fastjson.JSON;
import com.runa.monitor.comm.dto.Point;
import com.runa.monitor.comm.dto.PointValue;
import com.runa.monitor.platinterface.base.async.AsyncUnit;
import com.runa.monitor.platinterface.base.async.FutureDataTask;
import com.runa.monitor.platinterface.netservice.connect.AsyncCacheSession;
import com.runa.monitor.platinterface.netservice.connect.IRequest;
import com.runa.monitor.platinterface.netservice.connect.Session;
import com.runa.monitor.platinterface.netservice.connect.TellSession;
import com.runa.monitor.platinterface.netservice.connect.commstatus.CommStatus;
import com.runa.protocol.dto.*;
import com.runa.protocol.enums.ByteOrder;
import com.runa.protocol.enums.ModbusDataType;
import com.runa.protocol.enums.ModbusFunctionCode;
import com.runa.protocol.enums.ModbusType;
import com.runa.protocol.util.ByteUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.math.BigDecimal;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;

/**
 * 标准plc实现，以modbus协议为基准
 */
public abstract class PlcService implements AsyncUnit<Map> {

    static final Logger logger = LoggerFactory.getLogger(PlcService.class);

    /**
     * 点控请求优先级
     */
    static final int POINT_CTRL_PRIORITY = 0;

    /**
     * 点读请求优先级
     */
    static final int POINT_READ_PRIORITY = 1;

    /**
     * 全读请求优先级
     */
    static final int ALL_READ_PRIORITY = 2;


    private final Map<String, Object> rr = new HashMap();

    Integer increKey;

    /**
     * 该plc持有的所有点位
     */
    Map<Integer, IPoint> allPoints = new LinkedHashMap<>();

    List<IRequest<IRequestData>> allReadRequests = new ArrayList<>();

//    private List<IRequest<IRequestData>> pointReadRequests = newComm ArrayList<>();

    Map<IRequest<IRequestData>, List<IPoint>> requestPointTable = new HashMap<>();

    /**
     * 是否执行完成
     */
    boolean isReady = false;

    /**
     * 数据执行任务
     */
    FutureDataTask futureDataTask;

    /**
     * 主键
     */
    private final Integer id;

    /**
     * 从机地址
     */
    int slaveId;

    /**
     * modbus类型
     */
    private final ModbusType modbusType;

    /**
     * 通信链接会话
     */
    private Session session;

    /**
     * 单条指令的读取长度极限
     */
    Integer limit;

    private final int timeout;

    /**
     * 请求响应映射表
     */
    private final Map<Object, Object> requestResponse = new ConcurrentHashMap<>();

    PlcService(Integer id, int slaveId, Integer limit, Integer timeout, ModbusType modbusType, FutureDataTask task) {
        this.id = id;
        this.slaveId = slaveId;
        this.modbusType = modbusType;
        this.limit = limit == null ? 125 : limit;
        this.timeout = timeout == null ? 1000 : timeout;
        this.futureDataTask = task;
    }

    private List<ModbusPoint> pointConvector(List<Point> points) throws Exception {
        List<ModbusPoint> modbusPoints = new ArrayList<>();
        for (Point point : points) {
            modbusPoints.add(new ModbusPoint(point.getId(), point.getCode(), ModbusDataType.valuesOf(point.getDataType()),
                    Integer.valueOf(point.getAddress()), ByteOrder.valueOf(point.getByteOrder()), point.getOffset()));
        }
        return modbusPoints;
    }

    /**
     * 初始化
     *
     * @param points
     */
    public void init(List<Point> points) throws Exception {
        if (points != null && !points.isEmpty()) {
            List<ModbusPoint> mps = new ArrayList<>();
            pointConvector(points).forEach(point -> {
                mps.add(point);
                allPoints.put(point.getId(), point);
            });
            constructReadRequest(mps, allReadRequests, ALL_READ_PRIORITY, 0);
        }
    }

    private void constructReadRequest(List<ModbusPoint> points, List<IRequest<IRequestData>> collective, int ...flag) throws Exception {
        collective.clear();
        ModbusPoint prePoint = null;
        ModbusPoint startPoint = points.get(0);
        List<IPoint> sectionPoints = new ArrayList<>();
        for (ModbusPoint point : points) {
            if (prePoint != null) {
                int pc = point.getRegisterCount();
                //读取总长度小于限制长度
                boolean section = point.getAddress() + pc - startPoint.getAddress() >= limit;
                if (section) {
                    //读取长度等于前一个点位地址减去初始点位地址，再加上前点位的数据长度
                    int count = prePoint.getAddress() - startPoint.getAddress() + prePoint.getRegisterCount();
                    IRequest<IRequestData> request = createReq(new ModbusReadRequestData(slaveId, startPoint.readCode(), startPoint.getRegisterAddress(), count, modbusType));
                    request.flag(flag);
                    requestPointTable.put(request, sectionPoints);
                    collective.add(request);
                    request.collective(collective);
                    sectionPoints = new ArrayList<>();
                    //下一条指令从当前点位开始
                    startPoint = point;
                }
            }
            //最后一个点位
            if (point == points.get(points.size() - 1)) {
                int count = point.getAddress() - startPoint.getAddress() + point.getRegisterCount();
                IRequest<IRequestData> request = createReq(new ModbusReadRequestData(slaveId, startPoint.readCode(), startPoint.getRegisterAddress(), count, modbusType));
                request.flag(flag);
                requestPointTable.put(request, sectionPoints);
                collective.add(request);
                request.collective(collective);
            }
            prePoint = point;
            sectionPoints.add(prePoint);
        }
    }

    public abstract IRequest<IRequestData> createReq(IRequestData requestData);

    /**
     * 采集数据
     */
    public void collect(Integer increKey) {
        this.isReady = false;
        this.increKey = increKey;
        sendRequest(allReadRequests);
    }

    final AtomicLong queryStamp = new AtomicLong();

    public synchronized Map<String, Object> query(List<Integer> pointIds) {
        List<ModbusPoint> points = new ArrayList<>();
        for (Integer id : pointIds) {
            ModbusPoint point = (ModbusPoint) allPoints.get(id);
            if (point != null) {
                points.add(point);
            }
        }
        if (points.size() < pointIds.size()) {
            Map<String, Object> result = new HashMap<>();
            result.put("status", "查询失败，下列点位不存在: " + pointIds.stream().filter(
                    id -> !points.stream().map(ModbusPoint::getId)
                            .collect(Collectors.toList()).contains(id)).collect(Collectors.toList()));
            return result;
        }
        points.sort(Comparator.comparingInt(ModbusPoint::getAddress));
        List<IRequest<IRequestData>> pointReadRequests = new ArrayList<>();
        try {
            constructReadRequest(points, pointReadRequests, POINT_READ_PRIORITY, increment());
        } catch (Exception e) {
            Map<String, Object> result = new HashMap<>();
            result.put("status", e.getMessage());
            return result;
        }
        if (pointReadRequests.isEmpty()) {
            Map<String, Object> result = new HashMap<>();
            result.put("status", "查询失败，请求列表为空");
            return result;
        }
        Map<String, Object> values = null;
        sendRequest(pointReadRequests);
        queryStamp.set(System.currentTimeMillis());
        while (values == null) {
            if (System.currentTimeMillis() - queryStamp.get() > (long) timeout * 2 * pointReadRequests.size()) {
                values = new HashMap<>();
                values.put("status", "请求超时");
                logger.info("请求超时");
                break;
            }
            values = (Map<String, Object>) reqResp(pointReadRequests);
            try {
                Thread.sleep(1);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
        return values;
    }

    volatile String writeResult;

    final AtomicLong updateStamp = new AtomicLong();

    /**
     * 构造写指令请求并下发，每条指令写入的都是相邻点位
     *
     * @param pointValues
     */
    public synchronized Map<String, Object> constructAndSendWriteRequest(List<PointValue> pointValues) {
        List<ModbusPoint> points = new ArrayList<>();
        //从点位列表中获取要下发的点位
        for (PointValue pointValue : pointValues) {
            ModbusPoint point = (ModbusPoint) allPoints.get(pointValue.getId());
            if(point != null) {
                setPointValue(point, pointValue.getValue());
                points.add(point);
            }
        }
        Map<String, Object> result = new HashMap<>();
        if (pointValues.size() > points.size()) {
            result.put("status", "下发失败，下列点位不存在: " + pointValues.stream().filter(
                    pv -> !points.stream().map(ModbusPoint::getId)
                            .collect(Collectors.toList()).contains(pv.getId())).collect(Collectors.toList()));
            return result;
        }
        List<IRequest<IRequestData>> writeRequest = new ArrayList<>();
        pointsToRequest(writeRequest, points);
        if (writeRequest.isEmpty()) {
            result.put("status", "下发失败，请求列表为空");
            return result;
        }
        //发送请求
        sendRequest(writeRequest);
        updateStamp.set(System.currentTimeMillis());
        String status = null;
        while (status == null) {
            if (System.currentTimeMillis() - updateStamp.get() > (long) timeout * writeRequest.size()) {
                result.put("status", "请求超时");
                break;
            }
            status = writeResult;
        }
        writeResult = null;
        result.put("status", status);
        return result;
    }

    public synchronized Map<String, Object> generalWriteRequest(String address, String value, int type) {
        Map<String, Object> result = new HashMap<>();
        if (address.charAt(0) == '4') {
            int addr = Integer.parseInt(address.substring(1));
            byte[] bytes = null;
            if (type == 1) {
                bytes = value.getBytes();
            } else if (type == 2) {
                bytes = ByteUtil.hexStringToByte(value);
            } else if (type == 3) {
                bytes = value.getBytes();
                ByteUtil.reverseDoubleByte(bytes, 0, value.length());
            } else if (type == 4) {
                bytes = ByteUtil.hexStringToByte(value);
                ByteUtil.reverseDoubleByte(bytes, 0, value.length());
            }
            if (bytes != null) {
                List<IRequest<IRequestData>> writeRequest = new ArrayList<>();
                IRequest<IRequestData> request = createReq(new ModbusWriteRequestData(slaveId, ModbusFunctionCode.WRITE_REGISTERS, addr, modbusType, bytes));
                writeRequest.add(request);
                request.collective(writeRequest);
                request.flag(POINT_CTRL_PRIORITY, increment());
                sendRequest(writeRequest);
                updateStamp.set(System.currentTimeMillis());
                String status = null;
                while (status == null) {
                    if (System.currentTimeMillis() - updateStamp.get() > (long) timeout * writeRequest.size()) {
                        result.put("status", "请求超时");
                        break;
                    }
                    status = writeResult;
                }
                writeResult = null;
                result.put("status", status);
            }
        }
        return result;
    }

    private void pointsToRequest(List<IRequest<IRequestData>> requests, List<ModbusPoint> points) {
        if (points.isEmpty()) {
            return;
        }
        //根据modbus地址升序排列
        points.sort(Comparator.comparingInt(ModbusPoint::getAddress));
        ModbusPoint prePoint = null;
        List<ModbusPoint> mpnn = new ArrayList<>();
        List<List<ModbusPoint>> mpnnn = new ArrayList<>();
        //构造连续点位序列
        for (ModbusPoint point : points) {
            if (prePoint != null) {
                if (point.getAddress() - prePoint.getAddress()
                        > prePoint.getRegisterCount()) {
                    mpnnn.add(mpnn);
                    mpnn = new ArrayList<>();
                }
            }
            mpnn.add(point);
            prePoint = point;
        }
        mpnnn.add(mpnn);
        //每组连续的点位构造一个写请求
        int flag = increment();
        for (List<ModbusPoint> list : mpnnn) {
            if (list.isEmpty()) {
                continue;
            }
            ModbusDataType dataType = list.get(0).getDataType();
            int beginAddress = list.get(0).getRegisterAddress();
            IRequest<IRequestData> request = createReq(new ModbusWriteRequestData(slaveId,
                    ModbusDataType.Modbus_Bit.equals(dataType) ? ModbusFunctionCode.WRITE_COILS
                            : ModbusFunctionCode.WRITE_REGISTERS, beginAddress, list, modbusType));
            request.collective(requests);
            request.flag(POINT_CTRL_PRIORITY, flag);
            requests.add(request);
        }
    }

    void setPointValue(ModbusPoint point, String value) {
        Object object = null;
        switch (point.getDataType()) {
            case Modbus_Bit:
            case Modbus_Boolean:
                object = Boolean.valueOf(value);
                break;
            case Modbus_Int16:
            case Modbus_UInt16:
            case Modbus_Int32:
            case Modbus_UInt32:
            case Modbus_Float:
            case Modbus_Double:
                object = new BigDecimal(value);
                break;
            default:
        }
        point.constructValue(object);
    }

    void sendRequest(List<IRequest<IRequestData>> requests) {
        if (session != null) {
            logger.info("PLC {} --- 在线状态: {}", id, session.isOnline());
            if (!(requests == null || requests.isEmpty())) {
                if (session.isOnline()) {
                    if (session instanceof TellSession) {
                        requests.forEach(req -> ((TellSession) session).pushRequest(req));
                    } else if (session instanceof AsyncCacheSession) {
                        requests.forEach(req -> ((AsyncCacheSession) session).pushRequest(req));
                    }
                } else {
                    requests.get(0).cancel(CommStatus.OffLine);
                }
            } else {
                logger.error("PLC {} 请求列表为空!", id);
            }
        } else {
            logger.error("PLC {} Session Null ", id);
        }
    }

    private void modbusPointValue(List<IPoint> points, byte[] bytes) {
        ModbusPoint startPoint = (ModbusPoint) points.get(0);
        for (IPoint ref : points) {
            ModbusPoint point = (ModbusPoint) ref;
            int regIdx = point.getAddress() - startPoint.getAddress();
            if (ModbusDataType.Modbus_Bit.equals(point.getDataType())) {
                point.setValue(ByteUtil.byteToBit(bytes[regIdx / 8], regIdx % 8));
            } else {
                if (ModbusDataType.Modbus_Boolean.equals(point.getDataType())) {
                    try {
                        point.setValue(ByteUtil.byteToBoolean(bytes, regIdx << 1, point.getDeviation(), point.getByteOrder()));
                    } catch (Exception e) {
                        logger.error("", e);
                    }
                } else {
                    Object value = null;
                    switch (point.getDataType()) {
                        case Modbus_Int16:
                            value = ByteUtil.byteToShort(bytes, regIdx << 1, point.getByteOrder());
                            break;
                        case Modbus_UInt16:
                            value = ByteUtil.byteToUShort(bytes, regIdx << 1, point.getByteOrder());
                            break;
                        case Modbus_Int32:
                            value = ByteUtil.byteToInt(bytes, regIdx << 1, point.getByteOrder());
                            break;
                        case Modbus_UInt32:
                            value = ByteUtil.byteToUInt(bytes, regIdx << 1, point.getByteOrder());
                            break;
                        case Modbus_Float:
                            value = ByteUtil.byteToFloat(bytes, regIdx << 1, point.getByteOrder());
                            break;
                        case Modbus_Double:
                            value = ByteUtil.byteToDouble(bytes, regIdx << 1, point.getByteOrder());
                            break;
                        case Modbus_UTF8:
                            value = ByteUtil.byteToString(bytes, regIdx << 1, point.getByteOrder());
                        default:
                    }
                    point.setValue(value);
                }
            }
        }
    }

    /**
     * 读操作回调处理
     */
    public void handlerReadData(IRequest<IRequestData> request) {
        logger.info("handlerReadData request {}", JSON.toJSONString(request));
        ModbusResponseData data = (ModbusResponseData) request.getResponse().getResponseData();
        logger.info("handlerReadData data {}", JSON.toJSONString(data));
        byte[] bytes = data.getData();

        List<IPoint> points = requestPointTable.get(request);
        try {
            modbusPointValue(points, bytes);
            logger.info("points: {}", JSON.toJSONString(points));
            List<IRequest<IRequestData>> collective = request.collective();
            for (IRequest<IRequestData> req : collective) {
                if (!req.beReady()) {
                    return;
                }
            }
            logger.info("collective == allReadRequests: {}", collective == allReadRequests);
            if (collective == allReadRequests) {
                this.isReady = true;
                this.futureDataTask.successCall(this);
            } else {
                Map<Integer, String> values = new HashMap<>(16);
                collective.forEach(req ->
                        requestPointTable.get(req).forEach(point ->
                                values.put(point.getId(), String.valueOf(point.getValue()))));
                Map<String, Object> map = new HashMap<>(3);
                map.put("status", CommStatus.OK.getStatus());
                map.put("data", values);
                reqResp(collective, map);
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
        resetReadRequest(request, CommStatus.OK);
    }

    /**
     * 写操作回调处理
     */
    public void handlerWriteData(IRequest<IRequestData> request) {
        List<IRequest<IRequestData>> writeRequest = request.collective();
        for (IRequest<IRequestData> req : writeRequest) {
            if (!req.beReady()) {
                return;
            }
        }
        resetWriteRequest(CommStatus.OK);
    }

    abstract void timeoutOverLimit(CommStatus status);

    public void resetReadRequest(IRequest<IRequestData> request, CommStatus status) {
        if (request.collective() == allReadRequests) {
            //全读请求状态重置
            allReadRequests.forEach(IRequest::reset);
            if (!CommStatus.OK.equals(status)) {
                futureDataTask.failCall(this,
                        "Collector :" + this.id + "--" + status.getStatus());
            }
        } else {
            //删除请求-点位映射表中的点抄
            request.collective().forEach(req -> requestPointTable.remove(req));
            if (!CommStatus.OK.equals(status)) {
                Map<String, Object> map = new HashMap<>(1);
                map.put("status", status.getStatus());
                reqResp(request.collective(), map);
            }
        }
        timeoutOverLimit(status);
    }

    public void resetWriteRequest(CommStatus status) {
        if (status != null) {
            writeResult = status.getStatus();
        }
        timeoutOverLimit(status);
    }

    void reqResp(Object req, Object resp) {
        requestResponse.put(req, resp);
    }

    Object reqResp(Object req) {
        return requestResponse.remove(req);
    }

    public void updateCommStatus(boolean status) {
//        BaseContainer.netStatusCallback().plcStatusChange(id, subject, status);
    }

    public ModbusType getModbusType() {
        return modbusType;
    }

    protected void setSession(Session session) {
        this.session = session;
    }

    public Session session() {
        return session;
    }

    public Integer getId() {
        return id;
    }

    @Override
    public boolean ok() {
        return isReady;
    }

    public int getTimeout() {
        return timeout;
    }

    private final Map<Integer, String> msgData = new HashMap<>();

    @Override
    public Map get(AsyncState state) {
        switch (state) {
            case Correct:
                rr.put("plcId", id);
                rr.put("increKey", increKey);
                for (IPoint point : allPoints.values()) {
                    if (point.getValue() == null) {
                        msgData.put(point.getId(), null);
                    } else {
                        msgData.put(point.getId(), String.valueOf(point.getValue()));
                    }
                }
                rr.put("data", msgData);
                break;
            case Error:
                rr.put("plcId", id);
                rr.put("increKey", increKey);
                for (IPoint point : allPoints.values()) {
                    msgData.put(point.getId(), null);
                }
                rr.put("data", msgData);
                break;
            default:
                break;
        }
        return rr;
    }

    @Override
    public boolean equals(Object obj) {
        if (obj instanceof PlcService) {
            PlcService service = (PlcService) obj;
            return Objects.equals(service.getId(), id);
        }
        return false;
    }

    private final AtomicInteger increFlag = new AtomicInteger(1);

    int increment() {
        return increFlag.getAndIncrement();
    }

}
